package com.sf.shiva.oms.jstorm.topology;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sf.shiva.oms.jstorm.common.enumtype.SystemEnvEnum;
import com.sf.shiva.oms.jstorm.common.enumtype.TopologyNameEnum;
import com.sf.shiva.oms.jstorm.common.kafka.KafkaSpoutConfig;
import com.sf.shiva.oms.jstorm.common.kafka.KafkaSpoutFactory;
import com.sf.shiva.oms.jstorm.spout.CustomSchema;
import com.sf.shiva.oms.jstorm.bolt.SendKafkaBolt;
import com.sf.shiva.oms.jstorm.bolt.CalcBolt;
import com.sf.shiva.oms.jstorm.common.cfgmanage.CustomTopologyProperties;
import com.sf.shiva.oms.jstorm.common.cfgmanage.SystemProperties;
import com.sf.shiva.oms.jstorm.common.utils.SpringContext;
import com.sf.shiva.oms.jstorm.topology.CustomTopology;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;

/**
 * 
 * 描述：自定义拓扑结构
 *
 * <pre>
 * HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2018年4月2日      01369626         Create
 * ****************************************************************************
 * </pre>
 * 
 * @author 01369626
 * @since 1.0
 */
public class CustomTopology {

    private static final Logger logger = LoggerFactory.getLogger(CustomTopology.class);

    private CustomTopology() {

    }

    public static void main(String[] args) throws InvalidTopologyException {
        try {
            SystemProperties systemCfg = SpringContext.getInstance().getBean(SystemProperties.class);// 获取System配置信息
            CustomTopologyProperties topologyProps = SpringContext.getInstance().getBean(CustomTopologyProperties.class);// 获取拓扑配置信息
            if (SystemEnvEnum.DEV.getCode().equals(systemCfg.getSystemEnv())) {// 研发环境
                new LocalCluster().submitTopology(TopologyNameEnum.CUSTOM_TOPOLOGY.getName(), buildConfig(topologyProps), buildTopologyBuilder(topologyProps).createTopology());
            } else {
                StormSubmitter.submitTopology(TopologyNameEnum.CUSTOM_TOPOLOGY.getName(), buildConfig(topologyProps), buildTopologyBuilder(topologyProps).createTopology());
            }
        } catch (AlreadyAliveException e) {
            logger.error("customTopology starting error.", e);
        }
    }

    /**
     * 构建TopologyBuilder
     * 
     * @param topologyCfg
     *            topology配置信息
     * @return TopologyBuilder对象
     * @author 568677
     * @date 2017年12月4日
     */
    private static TopologyBuilder buildTopologyBuilder(CustomTopologyProperties topologyCfg) {
        TopologyBuilder builder = new TopologyBuilder();
        // 设置拓扑的spout和bolt以及数据流向
        // setSpout 方法的 第三个参数 3，就是Spout实例数，也是线程数，一个线程执行一个实例
        builder.setSpout("customKafkaSpout", KafkaSpoutFactory.create(SpringContext.getInstance().getBean("customKafkaSpout", KafkaSpoutConfig.class), new CustomSchema()), topologyCfg.getSpoutThreadNum());
        // setBolt
        // 第三个参数同上；setNumTasks方法原来是设置实例数量，不过在jstorm的后续版本中已不再使用，无效,jstorm随机分组方式推荐使用localOrShuffleGrouping,shuffle分组方式已不建议使用
        builder.setBolt("calcBolt", new CalcBolt(), topologyCfg.getCalBoltThreadNum()).localOrShuffleGrouping("customKafkaSpout");
        builder.setBolt("sendKafkaBolt", new SendKafkaBolt(), topologyCfg.getSendKafkaBoltThreadNum()).localOrShuffleGrouping("customBolt");
        return builder;
    }

    /**
     * 构建配置对象
     * 
     * @param props
     *            拓扑配置信息
     * @return 配置对象
     * @author 568677
     * @date 2017年12月4日
     */
    private static Config buildConfig(CustomTopologyProperties props) {
        Config config = new Config();
        config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, props.getMaxSpoutPending());// 一次读取KAFKA会读取多少数据
        config.put(Config.TOPOLOGY_WORKER_CHILDOPTS, props.getJavaOpts());// 自定义JAVA配置
        config.put("spout.single.thread", true);//保证kafka spout读取线程安全
        config.setNumWorkers(props.getWorkerNum());// jvm数量，本地模式无效
        config.setDebug(props.isDebugFlag());// 本地模式调试时开启
        config.setMessageTimeoutSecs(props.getMessageTimeoutSeconds());//消息处理超时时间
        return config;
    }
}
